package com.aotain.nyx.bots;

import org.apache.commons.codec.binary.Base64;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;

/**
 * 统计一个源IP访问的相同URL的个数的Map
 * @author Administrator
 *
 */
public class BotsStatMap  implements MapFunction<Tuple2<String,String>,Tuple2<String,BotsTuple>>{

	/**
	 * 
	 */
	private static final long serialVersionUID = -8707268481486415698L;


	@Override
	public Tuple2<String, BotsTuple> map(Tuple2<String, String> arg0)
			throws Exception {
		// TODO Auto-generated method stub
//		//dip+ "|" +dport+ "|" +sip+ "|" +sport+ "|" +gisstrBase64+ "|" 
//		//+ domainname+ "|" +url+ "|" +type+ "|" +delaytime+ "|" +idcno+ "|" 
//		// +upStreamOctets+ "|" +upStreamPacket+ "|" +status+ "|" +userAgent+ "|" +cookie+ "|" 
//		// +dnStreamOctets+ "|" +dnStreamPacket+ "|" +refer
		
		Tuple2<String, BotsTuple> ret = new Tuple2<String, BotsTuple>();
		String logString = arg0.f1; //格式化后的原始数据
		String[] items = logString.split("\\|",-1);
		if(items.length <= 10)
			return ret;
		String sip = arg0.f0;
		String dip = items[0];
		String dport = items[1];
		String gis = new String(Base64.decodeBase64(items[4]));
		String idc = items[9];
		String flowdir = items[19];
		long upStreamOctets = 0; //UPSTREAMOCTETS
		long upStreamPacket = 0; //UPSTREAMPACKET
//		String status = items[12];
//		String type = items[7];
//		long dnStreamOctets = 0; //DNSTREAMOCTETS
//		long dnStreamPacket = 0; //DNSTREAMPACKET
		
		String tripleKey = dip + "|" + sip + "|" + dport + "|" + gis + "|" + idc + "|" +flowdir;
		
		upStreamOctets = Long.parseLong(items[10]);
		upStreamPacket = Long.parseLong(items[11]);
//		dnStreamOctets = Long.parseLong(items[15]);
//		dnStreamPacket = Long.parseLong(items[16]);
		
		
			BotsTuple info = new BotsTuple();
			info.setUpStreamOctets(upStreamOctets);
			info.setUpStreamPacket(upStreamPacket);
			info.setSourceCount(1);
			String retKey = String.format("%s", tripleKey);
			
			ret.setFields(retKey, info);
		return ret;
	}

}
